package gupao.concurrency.service;

import gupao.concurrency.mock.TrafficMirrorMock;
import gupao.concurrency.model.TrafficRequest;
import gupao.concurrency.threadpool.Utils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
@Service
public class TrafficForecastService {

    final BlockingQueue<TrafficRequest> requestsQueue = new ArrayBlockingQueue<>(100000);
    AtomicBoolean ctl = new AtomicBoolean(false);
    AtomicBoolean shutdown = new AtomicBoolean(false);

    ExecutorService forecastServiceExecutor = Executors.newSingleThreadExecutor();
    Future<?> f;

    @PostConstruct
    public void init(){
        f = forecastServiceExecutor.submit(() -> {
            while(!shutdown.get()){
                try{
                    if(!ctl.get()){
//                        Thread.yield();
                        Utils.sleepIgnoreExp(1000);
                    } else {
                        TrafficRequest request = TrafficMirrorMock.nextRequest();
                        requestsQueue.offer(request, 100, TimeUnit.MILLISECONDS);
                    }
                }catch (Throwable t){
                    //ignore
                    log.error("forecast thread error", t);
                }
            }
        });

    }

    public void singleRequest(TrafficRequest req){
        log.info("forecast a single request: {}",req);
    }

    public void startForecast(){
        ctl.set(true);
    }

    public void stopAndSaveResult() {
        log.info("stopAndSaveResult");
    }

    public BlockingQueue<TrafficRequest> forecastTrafficQueue(){
        return requestsQueue;
    }

    public void stopForecast() {
        ctl.set(false);
    }

    public void shutdown(){
        shutdown.set(true);
        f.cancel(true);
    }
}
